package com.dc.dbus.transport.rabbitmq;

import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;

import com.dc.dbus.transport.commons.ClientSideReceiver;
import com.dc.dbus.transport.commons.ClientSideTransport;

public class RabbitMqClientSideTransport implements ClientSideTransport {
	
	protected final RabbitTemplate template;
	
	protected final RabbitMqManager rabbitMqManager;
	
	protected final String replyQueueName;
	
	protected final NoExceptionEventListener eventListener;
	
	public RabbitMqClientSideTransport(
			RabbitMqManager rabbitMqManager, EventListener eventListener) {
		super();
		this.template = rabbitMqManager.getTemplate();
		this.rabbitMqManager = rabbitMqManager;
		this.eventListener = new NoExceptionEventListener(eventListener);
		
		Queue reply = rabbitMqManager.getAdmin().declareQueue();
		String replyQueueName = reply.getName();
		this.replyQueueName = replyQueueName;
		
	}
	
	@Override
	public void startClientSideReceiver(final ClientSideReceiver clientSideReceiver) {
		
		MessageListenerAdapter listenerAdapter = new MessageListenerAdapter(new Object() {
			@SuppressWarnings("unused")
			public void handleMessage(Object param) {
				try {
					if (param instanceof MsgBack) {
						MsgBack msg = (MsgBack) param;
						clientSideReceiver.handlerMsg(msg.replyForMsgId, msg.replyForCmd, msg.resultCode, msg.returnObj);
					} else {
						eventListener.exception("unknow msg type from server, msg type:" + param.getClass().toString(), 
								new RuntimeException());
					}
				} catch (Exception e) {
					eventListener.exception("RabbitMqClientSideTansport.handleMsg", e);
				}
			}
		});
		
        SimpleMessageListenerContainer listener = new SimpleMessageListenerContainer(rabbitMqManager.getCf());
        listener.setMessageListener(listenerAdapter);
        listener.setQueueNames(replyQueueName);
        listener.setAcknowledgeMode(AcknowledgeMode.NONE);
        listener.setChannelTransacted(false);
        listener.start();
	}
	
	@Override
	public void send(String serviceName, String router, long msgId, int cmd, Object param, int invokeSessionTimeout) {
		
		int splitIndex = router.indexOf(',');
    	String exchange = router.substring(0, splitIndex);
    	String routingKey = router.substring(splitIndex+1, router.length());
		
    	Msg msg = new Msg(msgId, cmd, param, replyQueueName);
		template.convertAndSend(exchange, routingKey, msg);
	}
	
	
	
}
